package org.tao.lightningmq.nameserver.event.spi.listener;

import com.alibaba.fastjson2.JSON;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.AttributeKey;
import io.netty.util.internal.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.common.dto.BrokerMasterChangeReqDTO;
import org.tao.lightningmq.common.enums.BrokerEventCode;
import org.tao.lightningmq.common.enums.NameServerResponseCode;
import org.tao.lightningmq.common.enums.RegistryTypeEnum;
import org.tao.lightningmq.common.event.Listener;
import org.tao.lightningmq.common.tcp.TcpMsg;
import org.tao.lightningmq.nameserver.common.CommonCache;
import org.tao.lightningmq.nameserver.event.model.UnRegistryEvent;
import org.tao.lightningmq.nameserver.store.ServiceInstance;

import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.UUID;
import java.util.stream.Collectors;

/**
 * @Author lht
 * @date 2025/1/8 - 14:56
 * @description:
 */
public class UnRegistryListener implements Listener<UnRegistryEvent> {

    private static final Logger LOGGER = LoggerFactory.getLogger(UnRegistryListener.class);

    @Override
    public void onReceive(UnRegistryEvent event) throws Exception {
        ChannelHandlerContext context = event.getContext();
        Object reqId = context.attr(AttributeKey.valueOf("reqId")).get();
        if (reqId == null) {
            context.writeAndFlush(new TcpMsg(NameServerResponseCode.ERROR_USER_OR_PASSWORD,
                    NameServerResponseCode.ERROR_USER_OR_PASSWORD.getDesc().getBytes()));
            context.close();
            throw new IllegalAccessException("error account to connected!");
        }
        String reqIdStr = (String) reqId;
        ServiceInstance serviceInstance = CommonCache.getServiceInstanceManager().getServiceInstance(reqIdStr);

        // 判断下线的服务是否是broker节点，如果是而且是集群模式下的master节点，则需要选举新的master节点
        if (RegistryTypeEnum.BROKER.getCode().equals(serviceInstance.getRegistryType())) {
            selectNewMasterBroker(serviceInstance);
        }

        // 将服务实例从缓存中移除，同时关闭连接
        CommonCache.getServiceInstanceManager().removeServiceInstance(reqIdStr);
        LOGGER.info("unRegistry service:{}", serviceInstance);
        if (context.channel().isActive()) {
            context.writeAndFlush(new TcpMsg(NameServerResponseCode.UN_REGISTRY_SERVICE,
                    NameServerResponseCode.UN_REGISTRY_SERVICE.getDesc().getBytes()));
            context.close();
        }
    }

    private static void selectNewMasterBroker(ServiceInstance serviceInstance) {
        // TODO 待完善，在这之前需要先做完broker消费进度的同步
        Map<String, Object> attrs = serviceInstance.getAttrs();
        String brokerClusterRole = (String) attrs.getOrDefault("role", "");
        String brokerClusterGroup = (String) attrs.getOrDefault("group", "");
        boolean isClusterMode = !StringUtil.isNullOrEmpty(brokerClusterRole) && !StringUtil.isNullOrEmpty(brokerClusterGroup);
        if (isClusterMode && "master".equals(brokerClusterRole)) {
            LOGGER.warn("master node logout!");
            List<ServiceInstance> slaveInstances = CommonCache.getServiceInstanceManager().getServiceInstanceMap().values()
                    .stream()
                    .filter(node -> {
                        String group = (String) node.getAttrs().getOrDefault("group", "");
                        String role = (String) node.getAttrs().getOrDefault("role", "");
                        return group.equals(brokerClusterGroup) && "slave".equals(role);
                    }).collect(Collectors.toList());
//            long maxVersion = 0L;
//            ServiceInstance newMasterServiceInstance = null;
//            for (ServiceInstance slaveNode : slaveInstances) {
//                long latestVersion = (long) slaveNode.getAttrs().getOrDefault("latestVersion", 0L);
//                if (maxVersion <= latestVersion) {
//                    newMasterServiceInstance = slaveNode;
//                    maxVersion = latestVersion;
//                }
//            }
            int newSlaveIdx = new Random().nextInt(slaveInstances.size());
            ServiceInstance newMasterServiceInstance = slaveInstances.get(newSlaveIdx);
            if(newMasterServiceInstance != null) {
                newMasterServiceInstance.getAttrs().put("role", "master");
                LOGGER.info("new master node:{}", newMasterServiceInstance);
            }

            // 将主节点切换的消息发送给Broker
            String oldMasterAddr = serviceInstance.getIp() + ":" + serviceInstance.getPort();
            String newMasterAddr = newMasterServiceInstance.getIp() + ":" + newMasterServiceInstance.getPort();
            BrokerMasterChangeReqDTO reqDTO = new BrokerMasterChangeReqDTO(newMasterAddr, oldMasterAddr);

            for (ServiceInstance instance : slaveInstances) {
                if (instance == newMasterServiceInstance) continue;
                reqDTO.setMsgId(UUID.randomUUID().toString());
                TcpMsg tcpMsg = new TcpMsg(BrokerEventCode.BROKER_MASTER_CHANGE_MSG.getCode(),
                        JSON.toJSONBytes(reqDTO));
                instance.getContext().writeAndFlush(tcpMsg);
            }

        }
    }
}
